package com.shihan.rabbitmq.consumer;

import com.rabbitmq.client.*;
import com.shihan.rabbitmq.utils.ConnUtil;

import java.io.IOException;
import java.net.URISyntaxException;
import java.security.KeyManagementException;
import java.security.NoSuchAlgorithmException;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;

/**
 * @Description 消费端使用 QoS 进行限流
 * @Author shihan
 * @Date 2021/1/3 12:04
 * @Version 1.0
 */
public class QoSConsumer {

    public static final String QUEUE_NAME_QOS = "queue.qos.name";

    public static void main(String[] args) throws URISyntaxException, IOException, TimeoutException, NoSuchAlgorithmException, KeyManagementException {
        Channel channel = ConnUtil.getChannel();

        // 声明队列
        channel.queueDeclare(QUEUE_NAME_QOS,false,false,false,null);

        // 通过 basicQoS() 在消费端做限流，仅对推送模式生效
        // prefetchCount 表示消费端，能够处理的最大消息个数
        int prefetchCount = 10;
        // 常用方式一：
        channel.basicQos(prefetchCount);
        // 常用方式二：
        // 第二个参数 global 如果设置为 true，表示 QoS 设置，对使用当前 channel 的所有 consumer 都生效；
        // 第二个参数 global 如果设置为 false，表示 QoS 设置，仅对当前 consumer 生效； -- 在 DefaultConsumer 参数中设置 channel
        // channel.basicQos(prefetchCount,false);

        // 消费消息 -- 推送的方式
        channel.basicConsume(QUEUE_NAME_QOS,false,new DefaultConsumer(channel) {
            @Override
            public void handleDelivery(String consumerTag,
                                       Envelope envelope,
                                       AMQP.BasicProperties properties,
                                       byte[] body) throws IOException {
                //【疑问】：消息批量确认，如何拿到批量的消息信息？怎么区分每一个消息信息？
                //【回答】：正常操作即可，一个 body 对应一个 Message;
                System.out.println("【消费端】：" + new String(body,"utf-8"));

                //【疑问】：批量消息确认，一定是 prefetchCount 全部消费完成后，对最后一个 deliveryTag 进行确认吗？到底原理是什么？
                //【回答】：批量确认会由底层自动执行，即当判断到某个 deliveryTag 之前的所有消息，全都被确认完时，进行的一次批量确认；
                // 对消息进行批量确认，避免一条消息确认一遍所带来的网络开销；
                channel.basicAck(envelope.getDeliveryTag(), true);
            }
        });

        // 让 channel 和 connection 保持连接，才能让消息监听保持下来；
        try { TimeUnit.SECONDS.sleep(5); } catch (InterruptedException e) { e.printStackTrace(); }
        ConnUtil.closeConn();
    }

}
class ProducerForCurrConsumer {

    public static final String EXCHANGE_NAME_TEMP_QOS = "exc.direct.qos.name";

    public static final String QUEUE_NAME_TEMP_QOS = "queue.qos.name";

    public static final String BINDING_KEY_TEMP_QOS = "direct.qos.key";

    public static void main(String[] args) throws URISyntaxException, IOException, TimeoutException, NoSuchAlgorithmException, KeyManagementException {
        Channel channel = ConnUtil.getChannel();

        channel.queueDeclare(QUEUE_NAME_TEMP_QOS,false,false,false,null);

        channel.exchangeDeclare(EXCHANGE_NAME_TEMP_QOS, BuiltinExchangeType.DIRECT);

        channel.queueBind(QUEUE_NAME_TEMP_QOS,EXCHANGE_NAME_TEMP_QOS,BINDING_KEY_TEMP_QOS);

        for (int i = 0; i < 5; i++) {
            channel.basicPublish(EXCHANGE_NAME_TEMP_QOS,BINDING_KEY_TEMP_QOS,null,("测试 QoS，发送的消息" + i).getBytes("utf-8"));
        }

        ConnUtil.closeConn();
    }

}
